package com.bw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TrafficStats;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Date;

/**
 * 流量总览统计应用
 * 统计店铺访问、商品访问、转化等关键指标
 * - TrafficOverviewApp.java ：统计整体流量概览指标
 */
public class TrafficOverviewApp {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 读取Kafka的页面日志主题
        String pageTopic = "ods_traffic";
        String groupId = "traffic_overview_group";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId);
        DataStream<String> pageLogStream = env.addSource(kafkaSource);
        // 3. 转换为JSONObject并提取时间戳生成水位线
        DataStream<JSONObject> jsonObjStream = pageLogStream.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSON.parseObject(s);
            }
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>
                        forBoundedOutOfOrderness(Duration.ofSeconds(10))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObject, long l) {
                                return jsonObject.getLong("ts");
                            }
                        })
        );
        // 4. 按设备ID和是否为新访客分组，统计UV
        SingleOutputStreamOperator<Tuple4<String, String, Long, Long>> uvStatStream =
                jsonObjStream.map(new MapFunction<JSONObject, Tuple4<String, String, Long, Long>>() {
                    @Override
                    public Tuple4<String, String, Long, Long> map(JSONObject jsonObject) throws Exception {
                        String mid = jsonObject.getJSONObject("common").getString("mid");
                        String isNew = jsonObject.getJSONObject("common").getString("is_new");
                        String channel = jsonObject.getJSONObject("common").getString("ch");
                        String deviceType = channel.contains("pc") ? "pc" : "wireless";
                        long uvCount = 1L;
                        long newUvCount = "1".equals(isNew) ? 1L : 0L;
                        return new Tuple4<>(deviceType, mid, uvCount, newUvCount);
                    }
                }).keyBy(new KeySelector<Tuple4<String, String, Long, Long>, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(Tuple4<String, String, Long, Long> tuple) throws Exception {
                        return new Tuple2<>(tuple.f0, tuple.f1);
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                        .reduce(new ReduceFunction<Tuple4<String, String, Long, Long>>() {
                            @Override
                            public Tuple4<String, String, Long, Long> reduce(Tuple4<String, String, Long, Long> value1,
                                                                           Tuple4<String, String, Long, Long> value2) throws Exception {
                                // 对于同一设备，uvCount和newUvCount只保留1个
                                return new Tuple4<>(value1.f0, value1.f1, 1L, Math.max(value1.f3, value2.f3));
                            }
                        });

        // 5. 统计PV、商品PV、加购、收藏等指标
        SingleOutputStreamOperator<TrafficStats> trafficStatsStream = jsonObjStream.map(new MapFunction<JSONObject, TrafficStats>() {
            @Override
            public TrafficStats map(JSONObject jsonObject) throws Exception {
                String statsTime = DateFormatUtil.toYmdHms(jsonObject.getLong("ts"));                String statsType = "10s";
                long pvCount = 1L;
                long goodsPvCount = "good_detail".equals(jsonObject.getJSONObject("page").getString("page_id")) ? 1L : 0L;
                
                // 从actions字段中提取加购和收藏行为
                long cartAddCount = 0L;
                long favoritesCount = 0L;
                if (jsonObject.containsKey("actions")) {
                    for (Object action : jsonObject.getJSONArray("actions")) {
                        JSONObject actionObj = (JSONObject) action;
                        String actionType = actionObj.getString("action").toLowerCase();
                        if ("addcart".equals(actionType)) {
                            cartAddCount = 1L;
                        } else if ("favor".equals(actionType)) {
                            favoritesCount = 1L;
                        }
                    }
                }
                // 初始化其他指标为0
                TrafficStats stats = new TrafficStats();
                stats.setStatsTime(statsTime);
                stats.setStatsType(statsType);
                stats.setPvCount(pvCount);
                stats.setGoodsPvCount(goodsPvCount);
                stats.setCartAddCount(cartAddCount);
                stats.setFavoritesCount(favoritesCount);
                stats.setCreateTime(System.currentTimeMillis()/1000);
                return stats;
            }
        }).keyBy(new KeySelector<TrafficStats, String>() {
            @Override
            public String getKey(TrafficStats trafficStats) throws Exception {
                return trafficStats.getStatsTime() + trafficStats.getStatsType();
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficStats>() {
                    @Override
                    public TrafficStats reduce(TrafficStats value1, TrafficStats value2) throws Exception {
                        // 累加各项指标
                        value1.setPvCount(value1.getPvCount() + value2.getPvCount());
                        value1.setGoodsPvCount(value1.getGoodsPvCount() + value2.getGoodsPvCount());
                        value1.setCartAddCount(value1.getCartAddCount() + value2.getCartAddCount());
                        value1.setFavoritesCount(value1.getFavoritesCount() + value2.getFavoritesCount());
                        return value1;
                    }
                }, new WindowFunction<TrafficStats, TrafficStats, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TrafficStats> input,
                                      Collector<TrafficStats> out) throws Exception {
                        // 这里需要结合uv统计结果，计算完整的流量总览指标
                        TrafficStats stats = input.iterator().next();
                        // 注意：以下数据为模拟值
                        // 实际项目中，应：
                        // 1. 从uvStatStream中获取UV数据（包括总UV、新UV、PC端UV、无线端UV）
                        // 2. 从订单相关数据流中获取支付数据
                        // 3. 结合实际数据计算跳失率和转化率
                        stats.setUvCount(100L);  // 模拟UV数据
                        stats.setNewUvCount(30L); // 模拟新访客数据
                        stats.setBounceRate(0.35); // 模拟跳失率：35%
                        stats.setGoodsUvCount(80L); // 模拟商品访客数据
                        stats.setPaymentUserCount(20L); // 模拟支付买家数据
                        stats.setPaymentAmount(new BigDecimal(15000)); // 模拟支付金额数据
                        stats.setPaymentConversionRate(0.2); // 模拟支付转化率：20%
                        stats.setPcUvCount(40L); // 模拟PC端UV数据
                        stats.setWirelessUvCount(60L); // 模拟无线端UV数据
                        out.collect(stats);
                    }
                });

        // 6. 将结果写入ClickHouse
        trafficStatsStream.addSink(MyClickHouseUtil.getSinkFunction(
                "insert into dws_traffic_overview_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
        ));

        // 7. 执行作业
        env.execute("TrafficOverviewApp");
    }
}